package com.ehotting.mqbd.core.sync;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;


public class MqSyncFutureMapHelper<T extends java.io.Serializable> {
	
	private final Logger logger = LoggerFactory.getLogger(MqSyncFutureMapHelper.class);
	
	private final Map<String,SyncFuture<T>> cache = new ConcurrentHashMap<String,SyncFuture<T>>();
	
	private volatile boolean isRun = true;
	
	public void set(String key){
		SyncFuture<T> syn = new SyncFuture<T>();
		cache.put(key,syn);
	}
	
	public T get(String key,long timeout) throws InterruptedException{
		SyncFuture<T> syn= cache.get(key);
		if(syn!=null) {
			try {
				return syn.get(timeout);
			} catch (InterruptedException e) {
				cache.remove(key);
				logger.error(e.getMessage());
				throw e;
			}
		}
		return null;
	}
	
	public T get(String key) throws InterruptedException{
		return get( key,10) ;
	}
	
	public T waiteAndGet(String key) throws InterruptedException{
		return waiteAndGet(key,10);
	}
	
	public T waiteAndGet(String key,long timeout) throws InterruptedException{
		SyncFuture<T> syn = new SyncFuture<T>();
		cache.put(key,syn);
		try {
			T t = syn.get(timeout);
			return t;
		} catch (InterruptedException e) {
			cache.remove(key);
			throw e;
		}
	}
	
	public void putAndNotify(String key,T t){
		SyncFuture<T> syn= cache.get(key);
		if(syn!=null){
			syn.setResponse(t);
		}
	}
	
	
	/***
	 * 3个小时释放内存数据
	 */
	public void start(){
	    logger.info("Start MqSyncFutureMapHelper start.");
		Runnable run = new Runnable(){
			@Override
			public void run() {
				while(isRun){
					try {
						Thread.sleep(5000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					for (Entry<String, SyncFuture<T>> entry: cache.entrySet()) {
					    SyncFuture<T> value = entry.getValue();
					    if(System.currentTimeMillis()/1000-value.getBeginTime()/1000>2*60*60){
					    	value.setResponse(null);
					    }
					}
				}
			}
		};
		Thread thead = new Thread(run);
		thead.start();
		logger.info("Finish MqSyncFutureMapHelper start.");
	}
	
	
	public void stop(){
		logger.info("Start MqSyncFutureMapHelper stop.");
		isRun = false;
		logger.info("Finish MqSyncFutureMapHelper stop.");
	}
}
